package Service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.ctrip.framework.apollo.ConfigService;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import mappers.FateDataDao;
import mappers.FateDataDao;
import mappers.FateJobDao;
import domain.BaseResponse;
import entity.FateData;
import entity.FateJob;
import enums.BaseResponseEnum;
import proxy.FateProxy;
import utils.ApolloUtils;
import utils.CompressUtil;
import lombok.extern.slf4j.Slf4j;
import okhttp3.MediaType;
import okhttp3.MultipartBody;
import okhttp3.RequestBody;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

@Service
@Slf4j
public class FateService {
    @Resource
    private FateDataDao fateDataDao;

    @Resource
    private FateJobDao fateJobDao;


    /**
     * 将文件下载到tarName中,将其解压到fileName中,然后在fileName中追加数据,最后上传
     * @param job_id 要下载数据的jobId
     * @param role 要下载数据的role
     * @param party_id 要下载数据的party_id
     * @param component_name 要下载数据的component_name
     * @param tarName  指定下载数据压缩包名称
     * @param fileName 指定解压后csv文件名称以及上传的数据文件
     * @param head 上传数据是否有表头
     * @param partition 上传数据分区数目
     * @param tableName 上传数据表名
     * @param nameSpace 上传数据表命名空间
     * @param drop 是否覆盖上传
     * @param list 追加文件写入的数据
     * @return BaseResponse
     */
    public BaseResponse updateData(String job_id, String role, String party_id, String component_name, String tarName, String fileName, String head, String partition, String tableName, String nameSpace, String drop, ArrayList<String> list) {
        log.info("updateData tarName:{}, fileName:{}, nameSpace:{}, tableName:{}", tarName, fileName, nameSpace, tableName);

        // 下载数据文件
        if(StringUtils.isNotBlank(job_id)) {
            BaseResponse outputRes = downloadData(job_id, role, party_id, component_name, tarName, fileName);
            if (BaseResponseEnum.SUCCESS.getErrCode() != outputRes.getErrCode()) {
                return outputRes;
            }
        }

        // 数据追加写入文件
        BaseResponse dataRes = data2CSV(fileName, list);
        if (BaseResponseEnum.SUCCESS.getErrCode() != dataRes.getErrCode()) {
            return dataRes;
        }

        // 文件更新后上传Fate
        BaseResponse uploadRes = dataUpload(fileName, head, partition, tableName, nameSpace, drop);
        if (BaseResponseEnum.SUCCESS.getErrCode() != uploadRes.getErrCode()) {
            return uploadRes;
        }

        return uploadRes;
    }


    /**
     * 将数据追加写入到fileName中，如果文件不存在或直接生成fileName文件后写入
     * @param fileName 文件名称
     * @param list 写入的数据
     * @return BaseResponse
     */
    public BaseResponse data2CSV(String fileName, ArrayList<String> list){
        BufferedWriter writer = null;
        try {
            //需要写入的文件的路径
            String currentDir = System.getProperty("user.dir");
            String baseCsvDir = ConfigService.getAppConfig().getProperty(ApolloUtils.BASECSVDIR, ApolloUtils.BASECSVDIR_DEFAULT);

            // 防止文件命名错误处理，多加一层目录
            CompressUtil.createDirectory(currentDir + baseCsvDir ,fileName.split("\\.")[0]);
            fileName = fileName.split("\\.")[0]+"/"+fileName;
            String filePath = currentDir + baseCsvDir + fileName;
//            log.info("data2CSV filePath:{} data:{}",filePath,list);

            File file = new File(filePath);
            if (file.exists()) {
                // 文件存在
                writer = new BufferedWriter(new FileWriter(filePath, true));
            } else {
                // 文件不存在
                writer = new BufferedWriter(new FileWriter(filePath, true));
                writer.write("id");
                writer.newLine();
            }

            for (String str : list) {
                writer.write(str);
                writer.newLine();
            }

            return BaseResponse.success();
        } catch (IOException e) {
            log.error("data2CSV error",e);
            return BaseResponse.error();
        } finally {
            if (null != writer) {
                try{
                    writer.close();
                } catch (IOException e) {
                    log.error("writer close error",e);
                }
            }
        }

    }

    /**
     * 数据上传任务
     * @param fileName 上传的数据文件
     * @param head 上传数据是否有表头
     * @param partition 上传数据分区数目
     * @param tableName 上传数据表名
     * @param nameSpace 上传数据表命名空间
     * @param drop 是否覆盖上传
     */
    public BaseResponse dataUpload(String fileName, String head, String partition, String tableName, String nameSpace, String drop) {
        try{
            String currentDir = System.getProperty("user.dir");
            String baseCsvDir = ConfigService.getAppConfig().getProperty(ApolloUtils.BASECSVDIR, ApolloUtils.BASECSVDIR_DEFAULT);
            // 防止文件命名错误处理，多加一层目录
            String filePath = currentDir + baseCsvDir + fileName.split("\\.")[0]+"/"+fileName;
            File file = new File(filePath);
            RequestBody fileBody = RequestBody.create(MediaType.parse("text/csv"), file);
            JSONObject metaObject = new JSONObject();
            RequestBody requestBody = new MultipartBody.Builder()
                    .setType(MultipartBody.FORM)
                    .addFormDataPart("file", fileName, fileBody)
                    .addFormDataPart("head", head)
                    .addFormDataPart("partition", partition)
                    .addFormDataPart("table_name", tableName)
                    .addFormDataPart("namespace", nameSpace)
                    .addFormDataPart("drop", drop)
                    .build();

//            log.info("dataUpload: {},filePath: {}",JSONObject.toJSONString(requestBody),filePath);
            BaseResponse response = FateProxy.dataUpload(requestBody);
            if (BaseResponseEnum.SUCCESS.getErrCode()==response.getErrCode()){
                CompressUtil.deleteFile(filePath);
            }
            return response;
        } catch (Exception e) {
            log.error("dataUpload error",e);
            return BaseResponse.error();
        }
    }

    /**
     * 输出数据（最多仅能输出100条）
     * @param job_id 要输出数据的job_id
     * @param role  要输出数据的role
     * @param party_id 要输出数据的party_id
     * @param component_name 要输出数据的component_name
     * @return BaseResponse
     */
    public BaseResponse outputData(String job_id, String role, String party_id, String component_name) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("job_id", job_id);
        jsonObject.put("role", role);
        jsonObject.put("party_id", party_id);
        jsonObject.put("component_name", component_name);

        return FateProxy.outputData(JSONObject.toJSONString(jsonObject));
    }

    /**
     * 数据全部数据并下载压缩包后解压为csv文件
     * @param job_id 要输出数据的job_id
     * @param role  要输出数据的role
     * @param party_id 要输出数据的party_id
     * @param component_name 要输出数据的component_name
     * @param tarName 输出数据下载的压缩包名称
     * @param fileName 压缩包解压后的csv文件名称
     * @return BaseResponse
     */
    public BaseResponse downloadData(String job_id, String role, String party_id, String component_name, String tarName, String fileName) {
        try {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("job_id", job_id);
            jsonObject.put("role", role);
            jsonObject.put("party_id", party_id);
            jsonObject.put("component_name", component_name);

            String currentDir = System.getProperty("user.dir");
            String baseCsvDir = ConfigService.getAppConfig().getProperty(ApolloUtils.BASECSVDIR, ApolloUtils.BASECSVDIR_DEFAULT);
            String baseTarDir = ConfigService.getAppConfig().getProperty(ApolloUtils.BASETARDIR, ApolloUtils.BASETARDIR_DEFAULT);
            String tarFilePath = currentDir + baseTarDir + tarName;
            // tar目录
            CompressUtil.createDirectory(currentDir+baseTarDir,null);

            String outputDir = currentDir + baseCsvDir + fileName.split("\\.")[0];
            //csv目录
            CompressUtil.createDirectory(outputDir,null);

            String csvFilePath = outputDir +"/" + fileName;

//            log.info("downloadData: {}, tarFilePath: {}, csvFilePath: {}",JSONObject.toJSONString(jsonObject),tarFilePath,csvFilePath);

            return FateProxy.downloadData(JSONObject.toJSONString(jsonObject), tarFilePath, outputDir, csvFilePath);
        } catch (Exception e) {
            log.error("downloadData error",e);
            return BaseResponse.error();
        }
    }

    /**
     * 提交求交任务
     * @param hostName0 求交表名(tmp)
     * @param hostNameSpace0 求交表命名空间
     * @param hostName1 求交表名(test)
     * @param hostNameSpace1 求交表命名空间
     * @param guestName0 求交表名(tmp)
     * @param guestNameSpace0 求交表命名空间
     * @param guestName1 求交表名(his)
     * @param guestNameSpace1 求交表命名空间
     * @return BaseResponse
     */
    public BaseResponse jobSubmit(String hostName0, String hostNameSpace0,String hostName1, String hostNameSpace1, String guestName0, String guestNameSpace0, String guestName1, String guestNameSpace1) {
        try {
//            log.info("hostName: {}, hostNameSpace: {}, guestName: {}, guestNameSpace: {}",hostName0,hostNameSpace0,guestName1,guestNameSpace1);
            String requesBody = getDslAndConf(hostName0, hostNameSpace0,hostName1, hostNameSpace1, guestName0, guestNameSpace0, guestName1, guestNameSpace1);
            return FateProxy.jobSubmit(requesBody);
        } catch (Exception e) {
            log.error("jobSubmit error",e);
            return BaseResponse.error();
        }
    }

    /**
     * 查询任务状态
     * @param job_id 查询任务的 job_id
     * @return String 任务状态
     */
    public String jobQuery(String job_id) {
        try {
//            log.info("jobQuery: {}",job_id);
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("job_id", job_id);
            BaseResponse jobRes = FateProxy.jobQuery(JSONObject.toJSONString(jsonObject));
            JSONObject queryRes = (JSONObject) jobRes.getResult();
            JSONArray resData = queryRes.getJSONArray("data");
            JSONObject data = (JSONObject) resData.get(0);
            return data.getString("f_status");
        } catch (Exception e) {
            log.error("jobQuery error",e);
            return null;
        }
    }

    /**
     * 生成求交任务的Dsl和Conf配置
     * @param hostName0 求交表名(tmp)
     * @param hostNameSpace0 求交表命名空间
     * @param hostName1 求交表名(test)
     * @param hostNameSpace1 求交表命名空间
     * @param guestName0 求交表名(tmp)
     * @param guestNameSpace0 求交表命名空间
     * @param guestName1 求交表名(his)
     * @param guestNameSpace1 求交表命名空间
     * @return String(完整的Dsl和Conf配置)
     */
    public String getDslAndConf(String hostName0, String hostNameSpace0,String hostName1, String hostNameSpace1, String guestName0, String guestNameSpace0, String guestName1, String guestNameSpace1) {

        try {
            String jsonString = "{\n" +
                    "  \"dsl\": {\n" +
                    "    \"components\": {\n" +
                    "      \"reader_0\": {\n" +
                    "        \"module\": \"Reader\",\n" +
                    "        \"output\": {\n" +
                    "          \"data\": [\n" +
                    "            \"data\"\n" +
                    "          ]\n" +
                    "        }\n" +
                    "      },\n" +
                    "      \"reader_1\": {\n" +
                    "        \"module\": \"Reader\",\n" +
                    "        \"output\": {\n" +
                    "          \"data\": [\n" +
                    "            \"data\"\n" +
                    "          ]\n" +
                    "        }\n" +
                    "      },\n" +
                    "      \"union_0\": {\n" +
                    "        \"module\": \"Union\",\n" +
                    "        \"input\": {\n" +
                    "          \"data\": {\n" +
                    "            \"data\": [\n" +
                    "              \"reader_0.data\",\n" +
                    "              \"reader_1.data\"\n" +
                    "            ]\n" +
                    "          }\n" +
                    "        },\n" +
                    "        \"output\": {\n" +
                    "          \"data\": [\n" +
                    "            \"data\"\n" +
                    "          ]\n" +
                    "        }\n" +
                    "      },\n" +
                    "      \"intersection_0\": {\n" +
                    "        \"module\": \"Intersection\",\n" +
                    "        \"input\": {\n" +
                    "          \"data\": {\n" +
                    "            \"data\": [\n" +
                    "              \"union_0.data\"\n" +
                    "            ]\n" +
                    "          }\n" +
                    "        },\n" +
                    "        \"output\": {\n" +
                    "          \"data\": [\n" +
                    "            \"data\"\n" +
                    "          ]\n" +
                    "        }\n" +
                    "      }\n" +
                    "    }\n" +
                    "  },\n" +
                    "  \"runtime_conf\": {\n" +
                    "    \"dsl_version\": 2,\n" +
                    "    \"initiator\": {\n" +
                    "      \"role\": \"host\",\n" +
                    "      \"party_id\": 10000\n" +
                    "    },\n" +
                    "    \"role\": {\n" +
                    "      \"arbiter\": [\n" +
                    "        10000\n" +
                    "      ],\n" +
                    "      \"host\": [\n" +
                    "        10000\n" +
                    "      ],\n" +
                    "      \"guest\": [\n" +
                    "        9999\n" +
                    "      ]\n" +
                    "    },\n" +
                    "    \"component_parameters\": {\n" +
                    "      \"common\": {\n" +
                    "        \"intersection_0\": {\n" +
                    "          \"only_output_key\": true\n" +
                    "         }\n" +
                    "      },\n" +
                    "      \"role\": {\n" +
                    "        \"host\": {\n" +
                    "          \"0\": {\n" +
                    "            \"reader_0\": {\n" +
                    "              \"table\": {\n" +
                    "                \"name\": \"test_host\",\n" +
                    "                \"namespace\": \"host\"\n" +
                    "              }\n" +
                    "            },\n" +
                    "            \"reader_1\": {\n" +
                    "              \"table\": {\n" +
                    "                \"name\": \"test_host\",\n" +
                    "                \"namespace\": \"host\"\n" +
                    "              }\n" +
                    "            }\n" +
                    "          }\n" +
                    "        },\n" +
                    "        \"guest\": {\n" +
                    "          \"0\": {\n" +
                    "            \"reader_0\": {\n" +
                    "              \"table\": {\n" +
                    "                \"name\": \"test_guest\",\n" +
                    "                \"namespace\": \"guest\"\n" +
                    "              }\n" +
                    "            },\n" +
                    "            \"reader_1\": {\n" +
                    "              \"table\": {\n" +
                    "                \"name\": \"test_guest\",\n" +
                    "                \"namespace\": \"guest\"\n" +
                    "              }\n" +
                    "            }\n" +
                    "          }\n" +
                    "        }\n" +
                    "      }\n" +
                    "    }\n" +
                    "  }\n" +
                    "}";

            ObjectMapper objectMapper = new ObjectMapper();

            // 解析 JSON
            JsonNode jsonNode = objectMapper.readTree(jsonString);

            // 修改特定字段的值
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("host").get("0").get("reader_0").get("table")).put("name", hostName0);
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("host").get("0").get("reader_0").get("table")).put("namespace", hostNameSpace0);
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("host").get("0").get("reader_1").get("table")).put("name", hostName1);
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("host").get("0").get("reader_1").get("table")).put("namespace", hostNameSpace1);
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("guest").get("0").get("reader_0").get("table")).put("name", guestName0);
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("guest").get("0").get("reader_0").get("table")).put("namespace", guestNameSpace0);
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("guest").get("0").get("reader_1").get("table")).put("name", guestName1);
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("guest").get("0").get("reader_1").get("table")).put("namespace", guestNameSpace1);

            // 生成修改后的 JSON 字符串
            return objectMapper.writeValueAsString(jsonNode);
        } catch (Exception e) {
            log.error("getDslAndConf update error", e);
            return null;
        }
    }

    /**
     * 生成求交任务的Dsl和Conf配置(仅TMP与HIS表求交)
     * @param hostName 求交表名(tmp)
     * @param hostNameSpace 求交表命名空间
     * @param guestName 求交表名(his)
     * @param guestNameSpace 求交表命名空间
     * @return String(完整的Dsl和Conf配置)
     */
    public String getDslAndConf(String hostName, String hostNameSpace, String guestName, String guestNameSpace) {

        try {
            String jsonString = "{\n" +
                    "  \"dsl\": {\n" +
                    "    \"components\": {\n" +
                    "      \"reader_0\": {\n" +
                    "        \"module\": \"Reader\",\n" +
                    "        \"output\": {\n" +
                    "          \"data\": [\n" +
                    "            \"data\"\n" +
                    "          ]\n" +
                    "        }\n" +
                    "      },\n" +
                    "      \"intersection_0\": {\n" +
                    "        \"module\": \"Intersection\",\n" +
                    "        \"input\": {\n" +
                    "          \"data\": {\n" +
                    "            \"data\": [\n" +
                    "              \"reader_0.data\"\n" +
                    "            ]\n" +
                    "          }\n" +
                    "        },\n" +
                    "        \"output\": {\n" +
                    "          \"data\": [\n" +
                    "            \"data\"\n" +
                    "          ]\n" +
                    "        }\n" +
                    "      }\n" +
                    "    }\n" +
                    "  },\n" +
                    "  \"runtime_conf\": {\n" +
                    "    \"dsl_version\": 2,\n" +
                    "    \"initiator\": {\n" +
                    "      \"role\": \"host\",\n" +
                    "      \"party_id\": 10000\n" +
                    "    },\n" +
                    "    \"role\": {\n" +
                    "      \"arbiter\": [\n" +
                    "        10000\n" +
                    "      ],\n" +
                    "      \"host\": [\n" +
                    "        10000\n" +
                    "      ],\n" +
                    "      \"guest\": [\n" +
                    "        9999\n" +
                    "      ]\n" +
                    "    },\n" +
                    "    \"component_parameters\": {\n" +
                    "      \"common\": {\n" +
                    "        \"intersection_0\": {\n" +
                    "          \"only_output_key\": true\n" +
                    "         }\n" +
                    "      },\n" +
                    "      \"role\": {\n" +
                    "        \"host\": {\n" +
                    "          \"0\": {\n" +
                    "            \"reader_0\": {\n" +
                    "              \"table\": {\n" +
                    "                \"name\": \"test_host\",\n" +
                    "                \"namespace\": \"host\"\n" +
                    "              }\n" +
                    "            }\n" +
                    "          }\n" +
                    "        },\n" +
                    "        \"guest\": {\n" +
                    "          \"0\": {\n" +
                    "            \"reader_0\": {\n" +
                    "              \"table\": {\n" +
                    "                \"name\": \"test_guest\",\n" +
                    "                \"namespace\": \"guest\"\n" +
                    "              }\n" +
                    "            }\n" +
                    "          }\n" +
                    "        }\n" +
                    "      }\n" +
                    "    }\n" +
                    "  }\n" +
                    "}";

            ObjectMapper objectMapper = new ObjectMapper();

            // 解析 JSON
            JsonNode jsonNode = objectMapper.readTree(jsonString);

            // 修改特定字段的值
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("host").get("0").get("reader_0").get("table")).put("name", hostName);
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("host").get("0").get("reader_0").get("table")).put("namespace", hostNameSpace);
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("guest").get("0").get("reader_0").get("table")).put("name", guestName);
            ((ObjectNode) jsonNode.get("runtime_conf").get("component_parameters").get("role").get("guest").get("0").get("reader_0").get("table")).put("namespace", guestNameSpace);

            // 生成修改后的 JSON 字符串
            return objectMapper.writeValueAsString(jsonNode);
        } catch (Exception e) {
            log.error("getDslAndConf update error", e);
            return null;
        }
    }

    /**
     * 更新数据库中表的jobId
     * @param jobId     对应表的jobId
     * @param tableName 表名
     * @param nameSpace 表命名空间
     */
    public void fateUpdate(String jobId, String tableName, String nameSpace){
        try {
            FateJob fateJob = new FateJob();
            fateJob.setJobId(jobId);
            fateJob.setTableName(tableName);
            fateJob.setNameSpace(nameSpace);

            FateJob queryJob = fateJobDao.queryTableName(fateJob.getTableName());
            if (null == queryJob) {
                fateJobDao.insert(fateJob);
            } else {
                fateJobDao.update(fateJob);
            }
//            log.info("fateUpdate result:{}",result);
        }catch (Exception e){
            log.error("fateUpdate error",e);
        }
    }

    /**
     * 查询数据库中指定状态的数据
     * @param status 数据状态
     * @return List<FateData>
     */
    public List<FateData> queryStatus(String status){
        List<FateData> fateDataList = new ArrayList<>();
        try {
            fateDataList = fateDataDao.queryStatus(status);
        }catch (Exception e){
            log.error("queryStatus error",e);
        }
        return fateDataList;

    }
}
